package org.chen.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.collect.Lists;
import feign.FeignException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.chen.config.BatchStatsManager;
import org.chen.dto.TaskDetailDTO;
import org.chen.enu.BatchStatus;
import org.chen.enu.TaskStatus;
import org.chen.exception.ErrorContext;
import org.chen.exception.ServiceException;
import org.chen.model.CommentBatch;
import org.chen.mapper.CommentBatchMapper;
import org.chen.mapper.CommentTaskMapper;
import org.chen.model.CommentData;
import org.chen.model.CommentTask;
import org.chen.model.RetryMessage;
import org.chen.progress.BatchProgressDTO;
import org.chen.request.BatchCommentRequestDTO;
import org.chen.request.BatchUpdateRequest;
import org.chen.service.DistributedRateLimiter;
import org.chen.service.IBatchCommentService;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.net.MalformedURLException;
import java.net.URL;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;

@Service
@Slf4j
@Transactional(rollbackFor = Exception.class)
public class CommentBatchServiceImpl implements IBatchCommentService {
    @Autowired
    private CommentTaskExecutionService taskExecutionService;

    @Autowired
    private TaskCreationService taskCreationService;

    @Autowired
    private CommentBatchMapper batchMapper;

    @Autowired
    private CommentTaskMapper taskMapper;

    @Qualifier("commentTaskExecutor")
    @Autowired
    private ThreadPoolTaskExecutor commentTaskExecutor;

    /**
     * 生成唯一批次号
     */
    private String generateBatchNo() {
        return "BATCH" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"))
                + String.format("%04d", new Random().nextInt(10000));
    }

    /**
     * 异步创建批量评论任务
     */
    public String createBatchCommentAsync(BatchCommentRequestDTO request) {
        String batchNo = generateBatchNo();

        try {
            // 创建批次记录
            CommentBatch batch = taskCreationService.createBatchRecord(batchNo, request);
            batchMapper.insert(batch);

            // 创建任务记录并分批插入
            taskCreationService.createBatchTasks(batchNo, batch, request.getTargetUrls());

            // 触发异步处理
            commentTaskExecutor.execute(() -> taskExecutionService.processCommentBatch(batchNo));

            return batchNo;
        } catch (Exception e) {
            log.error("创建批量评论任务失败: {}", e.getMessage());
            throw new RuntimeException("创建批量评论任务失败", e);
        }
    }

    /**
     * 更新批次信息
     */
    @Override
    public void updateBatchInfo(List<String> batchNos, BatchUpdateRequest request) {
        for (String batchNo : batchNos) {
            try {
                // 获取并验证批次
                CommentBatch batch = getBatchOrThrow(batchNo);

                // 更新批次信息
                updateBatchBasicInfo(batch, request);

                if (request.getResubmit()) {
                    resubmitBatch(batch);
                } else {
                    batchMapper.updateBatchInfo(batch);
                }
            } catch (Exception e) {
                log.error("更新批次失败 {}: {}", batchNo, e.getMessage());
                throw new RuntimeException("更新批次失败: " + batchNo, e);
            }
        }
    }

    /**
     * 获取批次或抛出异常
     */
    private CommentBatch getBatchOrThrow(String batchNo) {
        CommentBatch batch = batchMapper.selectByBatchNo(batchNo);
        if (batch == null) {
            throw new RuntimeException("批次不存在: " + batchNo);
        }
        return batch;
    }

    /**
     * 更新批次基本信息
     */
    private void updateBatchBasicInfo(CommentBatch batch, BatchUpdateRequest request) {
        batch.setNickname(request.getNickname());
        batch.setEmail(request.getEmail());
        batch.setWebsiteUrl(request.getWebsiteUrl());
        batch.setCommentContent(request.getCommentContent());
        batch.setUpdateTime(LocalDateTime.now());
    }

    /**
     * 重新提交批次
     */
    private void resubmitBatch(CommentBatch batch) {
        // 重置批次状态
        batch.setSuccessCount(0);
        batch.setFailureCount(0);
        batch.setStatus(TaskStatus.PENDING.name());

        // 重置所有任务状态
        taskMapper.resetTasksForBatch(batch.getBatchNo(), batch);

        // 保存更新
        batchMapper.updateBatch(batch);

        // 重新触发异步处理
        commentTaskExecutor.execute(() -> taskExecutionService.processCommentBatch(batch.getBatchNo()));
    }

    /**
     * 恢复批次处理
     */
    @Override
    public void resumeBatch(String batchNo) {
        try {
            boolean exists = taskMapper.exists(
                    new QueryWrapper<CommentTask>()
                            .eq("batch_no", batchNo)
                            .eq("status", TaskStatus.PENDING.name())
            );

            if (exists) {
                resumeBatchProcessing(batchNo);
            }
        } catch (Exception e) {
            log.error("恢复批次失败 {}: {}", batchNo, e.getMessage());
            throw new RuntimeException("恢复批次失败: " + batchNo, e);
        }
    }

    /**
     * 恢复批次处理
     */
    private void resumeBatchProcessing(String batchNo) {
        // 更新批次信息
        batchMapper.updateBatchForResume(
                batchNo,
                BatchStatus.PROCESSING.name()
        );

        // 重新执行批量评论任务
        commentTaskExecutor.execute(() -> taskExecutionService.processCommentBatch(batchNo));
    }

    /**
     * 获取批次进度
     */
    public BatchProgressDTO getBatchProgress(String batchNo) {
        CommentBatch batch = getBatchOrThrow(batchNo);

        int completedCount = batch.getSuccessCount() + batch.getFailureCount();
        double progressPercentage = (double) completedCount / batch.getTotalCount() * 100;

        return BatchProgressDTO.builder()
                .batchNo(batchNo)
                .status(batch.getStatus())
                .totalCount(batch.getTotalCount())
                .successCount(batch.getSuccessCount())
                .failureCount(batch.getFailureCount())
                .progressPercentage(Math.round(progressPercentage * 100.0) / 100.0)
                .createTime(batch.getCreateTime())
                .updateTime(batch.getUpdateTime())
                .build();
    }

    /**
     * 获取批次详情
     */
    @Override
    public IPage<TaskDetailDTO> getBatchDetails(String batchNo, Integer current, Integer pageSize) {
        // 创建分页对象
        Page<CommentTask> page = new Page<>(current, pageSize);

        // 分页查询
        IPage<CommentTask> taskPage = taskMapper.selectPage(page,
                new QueryWrapper<CommentTask>()
                        .eq("batch_no", batchNo)
                        .orderByDesc("update_time"));

        // 转换为DTO
        return taskPage.convert(task -> TaskDetailDTO.builder()
                .targetUrl(task.getTargetUrl())
                .status(task.getStatus())
                .retryCount(task.getRetryCount())
                .errorMessage(task.getErrorMessage())
                .newUrl(task.getNewCommentUrl())
                .executeTime(task.getUpdateTime())
                .build());
    }
}